package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.KeywordStats;
import com.atguigu.gmall.realtime.utils.ClickhouseUtils;
import com.atguigu.gmall.realtime.utils.DateUtils;
import com.atguigu.gmall.realtime.utils.KafkaUtils;
import com.atguigu.gmall.realtime.utils.KeywordUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.List;

/**
 * 关键词主题统计(使用 StreamApi 实现)
 * <p>
 * 由于使用 flink sql 实现的时候，开窗聚合计算的地方存在问题，导致一直没有数据产生，因此用 StreamAPI 也实现了一下
 *
 * @author lvbingbing
 * @date 2022-07-06 19:05
 * @see KeywordStatsAppSql sql 方式实现
 */
@Slf4j
public class KeywordStatsAppStreamApi {
    public static void main(String[] args) throws Exception {
        // 1. 基本环境准备
        // 1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 1.2 并行度设置
        env.setParallelism(1);
        // 2. 检查点相关设置
        // 2.1 开启检查点
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置检查点超时时间
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        checkpointConfig.setCheckpointTimeout(60000L);
        // 2.3 开启外部化检查点，作业取消时保留检查点
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 6000L));
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/ck/gmall"));
        // 2.6 指定操作 hdfs 用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 3. 从页面主题中获取数据，筛选出搜索相关的数据，指定Watermark和事件时间字段，并做分组、开窗、聚合计算
        String topic = "dwd_page_log";
        String groupId = "keyword_stats_app_group";
        DataStreamSource<String> streamSource = env.addSource(KafkaUtils.getKafkaSource(topic, groupId));
        SingleOutputStreamOperator<KeywordStats> resDs = streamSource
                // 筛选出列表搜索相关的页面日志数据
                .filter(s -> {
                    JSONObject jsonObject = JSON.parseObject(s);
                    JSONObject pageJsonObj = jsonObject.getJSONObject("page");
                    String pageId = pageJsonObj.getString("page_id");
                    String item = pageJsonObj.getString("item");
                    return pageId.equals("good_list") && StringUtils.isNotEmpty(item);
                })
                // 筛选后进行分词
                .flatMap(new FlatMapFunction<String, String>() {
                    @Override
                    public void flatMap(String value, Collector<String> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject pageJsonObj = jsonObject.getJSONObject("page");
                        String item = pageJsonObj.getString("item");
                        List<String> keywordList = KeywordUtils.analyze(item);

                        Long ts = jsonObject.getLong("ts") / 1000;

                        for (String keyword : keywordList) {
                            JSONObject outJsonObject = new JSONObject();
                            outJsonObject.put("keyword", keyword);
                            outJsonObject.put("row_time", ts);
                            outJsonObject.put("ct", 1L);
                            String s = outJsonObject.toJSONString();
                            out.collect(s);
                        }
                    }
                })
                // 指定 Watermark 和事件时间字段
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                JSONObject jsonObject = JSON.parseObject(element);
                                return jsonObject.getLong("row_time") * 1000;
                            }
                        }))
                // 根据 keyword 分组
                .keyBy(s -> {
                    JSONObject jsonObject = JSON.parseObject(s);
                    return jsonObject.getString("keyword");
                })
                // 开一个 10s 的滚动事件时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 聚合结果
                .reduce(new ReduceFunction<String>() {
                    @Override
                    public String reduce(String value1, String value2) throws Exception {
                        JSONObject jsonObj1 = JSON.parseObject(value1);
                        JSONObject jsonObj2 = JSON.parseObject(value2);
                        Long count1 = jsonObj1.getLong("ct");
                        Long count2 = jsonObj2.getLong("ct");
                        jsonObj1.put("ct", count1 + count2);
                        return jsonObj1.toJSONString();
                    }
                }, new WindowFunction<String, KeywordStats, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<String> input, Collector<KeywordStats> out) throws Exception {
                        // 将聚合后的数据进行结构转换，向下游发送
                        for (String s1 : input) {
                            JSONObject jsonObject = JSON.parseObject(s1);
                            jsonObject.put("stt", DateUtils.formatDate(new Date(window.getStart())));
                            jsonObject.put("edt", DateUtils.formatDate(new Date(window.getEnd())));
                            jsonObject.put("source", "SEARCH");
                            jsonObject.put("ts", System.currentTimeMillis());
                            out.collect(JSON.parseObject(jsonObject.toJSONString(), KeywordStats.class));
                        }
                    }
                });
        // 4. 将结果写入到 clickhouse 中
        String executeSql = "insert into keyword_stats_2021(stt,edt,keyword,source,ct,ts) values(?,?,?,?,?,?)";
        resDs.addSink(ClickhouseUtils.operateClickhouseJdbcSink(executeSql));
        // 5. 触发程序执行
        env.execute();
    }
}
